// +build !aio

package net

import (
	"errors"
	"github.com/sniperHW/fly/core/buffer"
	"github.com/sniperHW/fly/core/queue"
	"net"
	"runtime"
	"sync/atomic"
	"time"
)

func isNetTimeout(err error) bool {
	switch err.(type) {
	case net.Error:
		if err.(net.Error).Timeout() {
			return true
		}
	default:
	}
	return false
}

type Socket struct {
	socketBase
	sendQue     *queue.ArrayQueue
	sendTimeout atomic.Value
	recvTimeout atomic.Value
}

func (this *Socket) SetRecvTimeout(timeout time.Duration) *Socket {
	this.recvTimeout.Store(timeout)
	return this
}

func (this *Socket) SetSendTimeout(timeout time.Duration) *Socket {
	this.sendTimeout.Store(timeout)
	return this
}

func (this *Socket) SetSendQueueSize(size int) *Socket {
	this.sendQue.SetCap(size)
	return this
}

func (this *Socket) getRecvTimeout() time.Duration {
	t := this.recvTimeout.Load()
	if nil == t {
		return 0
	} else {
		return t.(time.Duration)
	}
}

func (this *Socket) getSendTimeout() time.Duration {
	t := this.sendTimeout.Load()
	if nil == t {
		return 0
	} else {
		return t.(time.Duration)
	}
}

func (this *Socket) receiveAndUnpack() (interface{}, error) {
	var n int
	for {
		msg, err := this.inboundProcessor.Unpack()
		if nil != msg {
			return msg, nil
		} else if nil != err {
			return nil, err
		} else {
			timeout := this.getRecvTimeout()
			buff := this.inboundProcessor.GetRecvBuff()
			if len(buff) == 0 {
				panic("len(buff) == 0")
			}

			//fmt.Println("start recv", len(buff))

			if timeout > 0 {
				this.conn.SetReadDeadline(time.Now().Add(timeout))
				n, err = this.conn.Read(buff)
				this.conn.SetReadDeadline(time.Time{})
			} else {
				n, err = this.conn.Read(buff)
			}

			//fmt.Println("recv ok", n, err)

			if nil == err {
				this.inboundProcessor.OnData(buff[:n])
			} else {
				return nil, err
			}
		}
	}
}

func (this *Socket) recvThreadFunc() {
	defer this.ioWait.Done()
	for !this.testFlag(fclosed | frclosed) {
		var (
			p   interface{}
			err error
		)

		p, err = this.receiveAndUnpack()

		if !this.testFlag(fclosed | frclosed) {
			if nil != err {
				if isNetTimeout(err) {
					err = ErrRecvTimeout
				}
				if nil != this.errorCallback {
					if err != ErrRecvTimeout {
						this.setFlag(frclosed)
					}
					this.errorCallback(this, err)
				} else {
					this.Close(err, 0)
				}
			} else if p != nil {
				this.inboundCallBack(this, p)
			}
		} else {
			break
		}
	}
}

func (this *Socket) sendThreadFunc() {
	defer this.ioWait.Done()

	var err error

	localList := make([]interface{}, 0, 32)

	closed := false

	var i int

	var size int

	const maxsendsize = 65535

	var b *buffer.Buffer

	var offset int

	var n int

	defer func() {
		if nil != b {
			b.Free()
		}
	}()

	timeout := this.getSendTimeout()

	for {

		if i >= size {
			localList, closed = this.sendQue.Pop(localList)
			size = len(localList)
			if closed && size == 0 {
				break
			}
			i = 0
		}

		if nil == b {
			b = buffer.Get()
			for i < size {
				err = this.encoder.EnCode(localList[i], b)
				localList[i] = nil
				i++
				if nil != err {
					b.Free()
					b = nil
					if !this.testFlag(fclosed) {
						if nil != this.errorCallback {
							this.Close(err, 0)
							this.errorCallback(this, err)
						} else {
							this.Close(err, 0)
						}
					}
					return
				} else if b.Len() >= maxsendsize {
					break
				}
			}
			offset = 0
		}

		if timeout > 0 {
			this.conn.SetWriteDeadline(time.Now().Add(timeout))
			n, err = this.conn.Write(b.Bytes()[offset:])
			this.conn.SetWriteDeadline(time.Time{})
		} else {
			n, err = this.conn.Write(b.Bytes()[offset:])
		}

		offset += n

		if nil == err {
			b.Free()
			b = nil
		} else if !this.testFlag(fclosed) {
			if isNetTimeout(err) {
				err = ErrSendTimeout
			}

			if nil != this.errorCallback {
				if err != ErrSendTimeout {
					this.Close(err, 0)
				}
				this.errorCallback(this, err)
			} else {
				this.Close(err, 0)
			}

			if this.testFlag(fclosed) {
				return
			}
		} else {
			return
		}
	}
}

func (this *Socket) BeginRecv(cb func(*Socket, interface{})) (err error) {

	this.beginOnce.Do(func() {
		if nil == cb {
			panic("BeginRecv cb is nil")
		}

		if nil == this.inboundProcessor {
			panic("inboundProcessor is nil")
		}

		if this.testFlag(fclosed | frclosed) {
			err = ErrSocketClose
		} else {
			this.inboundCallBack = cb
			this.ioWait.Add(1)
			go this.recvThreadFunc()
		}
	})

	return
}

func (this *Socket) Send(o interface{}) (err error) {
	if this.testFlag(fclosed) {
		err = ErrSocketClose
	} else {
		if nil == o {
			panic("here")
		}
		err = this.sendQue.Append(o)
		if err == queue.ErrQueueClosed {
			err = ErrSocketClose
		} else if err == queue.ErrQueueFull {
			err = ErrSendQueFull
		} else {
			this.sendOnce.Do(func() {
				this.ioWait.Add(1)
				go this.sendThreadFunc()
			})
		}
	}
	return
}

func (this *Socket) Close(reason error, delay time.Duration) {

	this.closeOnce.Do(func() {
		runtime.SetFinalizer(this, nil)

		this.setFlag(fclosed)

		this.sendQue.Close()

		if this.sendQue.Len() == 0 {
			delay = 0
		} else {
			delay = delay * time.Second
		}

		if delay > 0 {
			this.ShutdownRead()
			ticker := time.NewTicker(delay)
			go func() {
				/*
				 *	delay > 0,sendThread最多需要经过delay秒之后才会结束，
				 *	为了避免阻塞调用Close的goroutine,启动一个新的goroutine在chan上等待事件
				 */
				select {
				case <-this.sendCloseChan:
				case <-ticker.C:
				}
				ticker.Stop()
				this.conn.Close()
			}()

		} else {
			this.conn.Close()
		}

		go func() {
			this.ioWait.Wait()
			if nil != this.closeCallBack {
				this.closeCallBack(this, reason)
			}
		}()

	})

}

func NewSocket(conn net.Conn) *Socket {

	switch conn.(type) {
	case *net.TCPConn, *net.UnixConn:
		break
	default:
		return nil
	}

	s := &Socket{
		socketBase: socketBase{
			conn:          conn,
			sendCloseChan: make(chan struct{}),
		},
		sendQue: queue.NewArrayQueue(128),
	}

	runtime.SetFinalizer(s, func(s *Socket) {
		s.Close(errors.New("gc"), 0)
	})

	return s
}

func CreateSocket(conn net.Conn) *Socket {
	return NewSocket(conn)
}
